package com.wzw.basics.concurrent.blockingqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 生产者消费者模型为例模式说明
 * <p>
 * 线程一生产，线程二消费
 * <p>
 * 多线程环境下，不能用if判断条件，应该用while循环或者for。
 *
 * @author Wangzhiwen
 */
public class ProducerConsumer {

    public static void main(String[] args) {
        // synchronized
        VersionOne versionOne = new VersionOne();
        moreThreadProducer(10, versionOne);
        moreThreadConsumer(10, versionOne);

        // lock condition
        VersionTwo versionTwo = new VersionTwo();
        moreThreadProducer(10, versionTwo);
        moreThreadConsumer(10, versionTwo);

        // lockSupport ..

        // blockingQueue
        VersionFour versionFour = new VersionFour();
        versionFour.setBlockingQueue(new ArrayBlockingQueue<>(10));
        moreThreadProducer(10, versionFour);
        moreThreadConsumer(10, versionFour);
    }

    public static void moreThreadProducer(int count, Version version) {
        for (int i = 1; i <= count; i++) {
            new Thread(version::producer, "Thread A " + i).start();
        }
    }

    public static void moreThreadConsumer(int count, Version version) {
        for (int i = 1; i <= count; i++) {
            new Thread(version::consumer, "Thread B " + i).start();
        }
    }

}

abstract class Version {
    protected int number;

    public abstract void producer();

    public abstract void consumer();
}

/**
 * synchronized wait notify
 * 缺点：
 * wait 等待
 * notify 唤醒，无法唤醒指定的线程
 * 在分组条件下，没有办法做到准确唤醒某一个线程。例如：线程A打印5次 然后线程B打印10次 然后线程C打印15次 在循环到线程A。实现就会比较复杂
 */
class VersionOne extends Version {

    private final Object lock = new Object();

    @Override
    public void producer() {
        // 同步
        synchronized (lock) {
            try {
                // 判断
                while (number != 0) {
                    lock.wait();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            // 执行
            number++;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            // 通知唤醒
            lock.notifyAll();
        }

    }

    @Override
    public void consumer() {
        synchronized (lock) {
            // 判断
            try {
                while (number == 0) {
                    lock.wait();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            number--;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            lock.notifyAll();
        }
    }
}

/**
 * condition await signal
 * 优点：
 * 条件可以分组，
 */
class VersionTwo extends Version {

    private final Lock lock = new ReentrantLock();

    private final Condition condition = lock.newCondition();

    @Override
    public void producer() {
        // 同步
        lock.lock(); // condition 一定要在lock锁内使用，否则会报错。

        try {

            // 判断标识，当前生产者是否可以生产
            while (number != 0) {
                // 等待
                condition.await();
            }

            // 处理逻辑
            number++;

            System.out.println(Thread.currentThread().getName() + "\t" + number);

            // 通知唤醒消费者
            condition.signalAll();

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

    @Override
    public void consumer() {
        lock.lock();

        try {

            // 判断消费者，能否消费
            while (number == 0) {
                condition.await();
            }

            // 消费
            number--;

            System.out.println(Thread.currentThread().getName() + "\t" + number);

            // 唤醒生产者
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

/**
 * LockSupport.park() LockSupport.unpark()
 */
class VersionThree extends Version {

    @Override
    public void producer() {
        try {
            while (number != 0) {
                LockSupport.park();
            }
            number++;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            LockSupport.unpark(Thread.currentThread());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public void consumer() {
        try {
            while (number == 0) {
                LockSupport.park();
            }
            number--;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            LockSupport.unpark(Thread.currentThread());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/**
 * BlockingQueue
 */
class VersionFour extends Version {
    // 生产和消费的标识，使用volatile修饰保证多线程间的可见性
    private volatile boolean flag = true;
    // 原子生产与消费，保证线程安全
    private final AtomicInteger atomic = new AtomicInteger();
    // 阻塞队列
    private BlockingQueue<Integer> blockingQueue;

    public void setBlockingQueue(BlockingQueue<Integer> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public void stop() {
        this.flag = false;
    }

    @Override
    public void producer() {
        // 中间变量建议放在循环体外
        int data;
        boolean isOffer = false;
        // 是否生产
        while (flag) {
            data = atomic.incrementAndGet();
            // 放入阻塞队列内
            try {
                isOffer = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 是否插入成功
            if (isOffer) {
                System.out.println(Thread.currentThread().getName() + " 插入队列：" + data + " 成功");
            } else {
                System.out.println(Thread.currentThread().getName() + " 插入队列：" + data + " 失败");
            }
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println(Thread.currentThread().getName() + "FLAG=false 停止生产");
    }

    @Override
    public void consumer() {
        Integer data = null;
        while (flag) {
            try {
                data = blockingQueue.poll(2L, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (data != null) {
                System.out.println(Thread.currentThread().getName() + " 消费队列：" + data + " 成功");
            } else {
                System.out.println(Thread.currentThread().getName() + " 消费队列失败，队列以为空");
                flag = false;
                return;
            }
        }
    }
}

/**
 * 多个线程间相互指定唤醒通知
 */
class ConditionGroup {

    private static volatile int flag = 1;

    private static final Lock lock = new ReentrantLock();

    private static final Condition conditionA = lock.newCondition();
    private static final Condition conditionB = lock.newCondition();
    private static final Condition conditionC = lock.newCondition();

    //线程A打印A 1次
    public static void m1() {
        try {
            // 同步
            lock.lock();
            // 判断
            while (flag != 1) {
                conditionA.await();
            }
            // 执行逻辑
            System.out.println(Thread.currentThread().getName() + " A");
            // 改变标识，唤醒线程B
            flag = 2;
            conditionB.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

    //线程B打印B 2次
    public static void m2() {
        try {
            lock.lock();
            while (flag != 2) {
                conditionB.await();
            }
            for (int i = 0; i < 2; i++) {
                System.out.println(Thread.currentThread().getName() + " B");
            }
            flag = 3;
            conditionC.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    //线程C打印C 3次 紧接着循环到线程A
    public static void m3() {
        try {
            lock.lock();
            while (flag != 3) {
                conditionC.await();
            }
            for (int i = 0; i < 3; i++) {
                System.out.println(Thread.currentThread().getName() + " C");
            }
            flag = 1;
            conditionA.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        new Thread(ConditionGroup::m1, "Thread A").start();
        new Thread(ConditionGroup::m2, "Thread B").start();
        new Thread(ConditionGroup::m3, "Thread C").start();
    }
}

